package com.atguigu.gmall.publisher.service.impl;

import com.atguigu.gmall.publisher.bean.DauInfo;
import com.atguigu.gmall.publisher.service.VisitRealtimeService;
import org.apache.commons.lang3.time.DateUtils;
import org.elasticsearch.ElasticsearchStatusException;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.rest.RestStatus;
import org.elasticsearch.search.aggregations.Aggregation;
import org.elasticsearch.search.aggregations.AggregationBuilders;
import org.elasticsearch.search.aggregations.bucket.terms.ParsedStringTerms;
import org.elasticsearch.search.aggregations.bucket.terms.Terms;
import org.elasticsearch.search.aggregations.bucket.terms.TermsAggregationBuilder;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;


@Service
public class VisitRealtimeServiceImpl implements VisitRealtimeService {


    public static final String  DAU_INDEX_PREFIX="gmall0819_dau_info_";

    @Autowired
    RestHighLevelClient restHighLevelClient;

    @Override
    public DauInfo getDauInfo(String td) {
        DauInfo dauInfo = new DauInfo();
        Long dauTotal = getDauTotal(td);
        dauInfo.setDauTotal(dauTotal);

        //今天 日活分时明细
        Map<String, Long> dauHourMapToday = getDauHourMap(td);
        dauInfo.setDauTd(dauHourMapToday);

        //昨天
        String yd = getYd(td);
        Map<String, Long> dauHourMapYesterday = getDauHourMap(yd);
        dauInfo.setDauYd(dauHourMapYesterday);

        return dauInfo;
    }


    private  String  getYd(String td){
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
        try {
            Date todayDate = dateFormat.parse(td);
            Date ydDate = DateUtils.addDays(todayDate, -1);
            String yd = dateFormat.format(ydDate);
            return yd;
        } catch (ParseException e) {
            e.printStackTrace();
            throw  new RuntimeException("日期格式有误");
        }

    }

    // 查询总数  //es
    private  Long  getDauTotal(String td){

        String  dauIndex=DAU_INDEX_PREFIX+td;
        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(dauIndex);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.size(0);
        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            long total = searchResponse.getHits().getTotalHits().value;
            return total;
        } catch (IOException e) {
            e.printStackTrace();
            throw  new RuntimeException("访问es 异常");
        }catch (ElasticsearchStatusException ese){
            if( ese.status()== RestStatus.NOT_FOUND  ){
                return  0L;
            }
        }
        return  0L;
    }


    //查询分时明细
    private Map<String,Long> getDauHourMap(String date){
        String  dauIndex=DAU_INDEX_PREFIX+date;

        SearchRequest searchRequest = new SearchRequest();
        searchRequest.indices(dauIndex);

        //查询条件
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.size(0);
        // 聚合部分
        TermsAggregationBuilder termsAggregationBuilder = AggregationBuilders.terms("groupby_hr").field("hr").size(24);
        searchSourceBuilder.aggregation(termsAggregationBuilder);

        searchRequest.source(searchSourceBuilder);
        try {
            SearchResponse searchResponse = restHighLevelClient.search(searchRequest, RequestOptions.DEFAULT);
            ParsedStringTerms groupby_hr =(ParsedStringTerms) searchResponse.getAggregations().get("groupby_hr");
            List<? extends Terms.Bucket> buckets = groupby_hr.getBuckets();
            HashMap<String, Long>  dauHrMap = new HashMap<>();
            for (Terms.Bucket bucket : buckets) {
                dauHrMap.put( bucket.getKey().toString(),bucket.getDocCount());
            }
            return dauHrMap;
        } catch (IOException e) {
            e.printStackTrace();
            throw  new RuntimeException("访问es 异常");
        }catch (ElasticsearchStatusException ese){
            if( ese.status()== RestStatus.NOT_FOUND  ){
                return  new HashMap<>();
            }
        }
        return  new HashMap<>();

    }

}
